package test.threads.queue.stressed;

import java.util.concurrent.Executor;

import com.lmax.disruptor.EventFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import test.helpers.Config;
import test.threads.queue.common.IUsefulWorkEmulator;

/**
 * Just process packets in calling thread
 *
 * @author cheremin
 * @since 01.03.12,  16:28
 */
public class SingleThreadedTask<E> extends AbstractTask<E> {
    private static final Log log = LogFactory.getLog( SingleThreadedTask.class );

    public SingleThreadedTask() {
    }

    public SingleThreadedTask( final int bufferSize,
                               final int packetsInBatch ) {
        super( bufferSize, packetsInBatch );
    }

    public SingleThreadedTask( final Config config ) {
        super( config );
    }

    @Override
    public int workersRequired() {
        // this is fake: we do not actually use 1 additional thread, instead we do all work in
        // calling thread. So we really use 1 worker, but it is calling thread worker,
        // not additional thread worker, as usually
        return 1;
    }

    private IUsefulWorkEmulator<E> emulator;

    @Override
    protected EventEnqueuer _initialize( final Executor threadPool,
                                         final IUsefulWorkEmulator<E>[] emulators,
                                         final EventFactory<E> factory ) throws Exception {
        this.emulator = emulators[0];
        return new EventEnqueuer<E>( eventsInBatch(), factory ) {
            @Override
            protected void enqueueEventsBatch( final int eventsInBatch,
                                               final EventFactory<E> factory ) throws Exception {
                //do nothing
            }
        };
    }

    @Override
    protected BenchmarkResult[] executeImpl( final long millisToBogus,
                                             final long millisToMeasure,
                                             final int measurements ) throws Exception {
        long eventsDequeued = 0;


//        final long msToOneMeasure = millisToMeasure / measurements;
        long eventsAtStart = eventsDequeued;
        long startedAtMs = System.currentTimeMillis();
        long startedAtNs = System.nanoTime();
        while ( true ) {
            dequeueBatchEvents( eventsInBatch() );
            eventsDequeued += eventsInBatch();

            final long finishedAtMs = System.currentTimeMillis();
            final long ellapsedMs = finishedAtMs - startedAtMs;
            if ( ellapsedMs >= millisToMeasure ) {
                final long eventsProcessed = eventsDequeued - eventsAtStart;
                final long finishedAtNs = System.nanoTime();
                final long ellapsedNs = finishedAtNs - startedAtNs;

                final BenchmarkResult result = new BenchmarkResult(
                        ellapsedMs,
                        ellapsedNs,
                        eventsProcessed
                );
                return new BenchmarkResult[]{ result };
            }
        }
    }

    @Override
    protected long dequeueBatchEvents( final int eventsInBatch ) throws Exception {
        final EventFactory<E> factory = eventFactory();
        for ( int i = 0; i < eventsInBatch; i++ ) {
            final E event = factory.newInstance();
            emulator.spendCPUTime( event );
        }
        return 0;
    }
}
